package com.bw.medical.realtime.common.util;


import com.bw.medical.realtime.common.constant.Constant;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.SimpleStringSerializer;

import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author 杨涵
 * @date 2025/3/7 13:28
 */
public class DorisSinkManager {
    // 全局缓存，每个表名对应一个Sink实例
    private static final Map<String, DorisSink<String>> SINK_CACHE = new ConcurrentHashMap<>();

    public static DorisSink<String> getDorisSink(String tableName) {
        // 先从缓存取
        DorisSink<String> sink = SINK_CACHE.get(tableName);
        if (sink == null) {
            synchronized (DorisSinkManager.class) {
                sink = SINK_CACHE.get(tableName);
                if (sink == null) {
                    // 第一次创建
                    sink = createDorisSink(tableName);
                    SINK_CACHE.put(tableName, sink);
                }
            }
        }
        return sink;
    }

    private static DorisSink<String> createDorisSink(String tableName) {
        Properties properties = new Properties();
        properties.setProperty("format", "json");
        properties.setProperty("read_json_by_line", "true");

        return DorisSink.<String>builder()
                .setDorisReadOptions(DorisReadOptions.builder().build())
                .setDorisExecutionOptions(DorisExecutionOptions.builder()
                        .setLabelPrefix("label-doris-" + tableName + "-" + System.currentTimeMillis())
                        .setDeletable(false)
                        .setStreamLoadProp(properties)
                        .build())
                .setSerializer(new SimpleStringSerializer())
                .setDorisOptions(DorisOptions.builder()
                        .setFenodes(Constant.FENODES)
                        .setTableIdentifier(Constant.DORIS_DATABASE + "." + tableName)
                        .setUsername(Constant.DORIS_USERNAME)
                        .setPassword(Constant.DORIS_PASSWORD)
                        .build())
                .build();
    }
}

